<!DOCTYPE html>



  


<html class="theme-next pisces use-motion" lang="zh-Hans">
<head><meta name="generator" content="Hexo 3.8.0">
  <meta charset="UTF-8">
<meta http-equiv="X-UA-Compatible" content="IE=edge">
<meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1">
<meta name="theme-color" content="#222">









<meta http-equiv="Cache-Control" content="no-transform">
<meta http-equiv="Cache-Control" content="no-siteapp">
















  
  
  <link href="/lib/fancybox/source/jquery.fancybox.css?v=2.1.5" rel="stylesheet" type="text/css">







<link href="/lib/font-awesome/css/font-awesome.min.css?v=4.6.2" rel="stylesheet" type="text/css">

<link href="/css/main.css?v=5.1.4" rel="stylesheet" type="text/css">


  <link rel="apple-touch-icon" sizes="180x180" href="/images/apple-touch-icon-next.png?v=5.1.4">


  <link rel="icon" type="image/png" sizes="32x32" href="/images/saiki.jpeg?v=5.1.4">


  <link rel="icon" type="image/png" sizes="16x16" href="/images/saiki.jpeg?v=5.1.4">


  <link rel="mask-icon" href="/images/logo.svg?v=5.1.4" color="#222">





  <meta name="keywords" content="Java,mina,">










<meta name="description" content="Java的事件机制一般都是通过观察者模式实现的(?).观察者模式就必须有观察者和被观察者.Mina里里面,再一开始acceptor = new NioSocketAcceptor(NetConfig.MINA_PROCESSOR);就在上面的父类AbstractIoService中注册了观察者:1234567891011121314151617181920212223242526272829303">
<meta name="keywords" content="Java,mina">
<meta property="og:type" content="article">
<meta property="og:title" content="Mina中的观察者模式">
<meta property="og:url" content="http://yoursite.com/2019/05/28/Mina中的观察者模式/index.html">
<meta property="og:site_name" content="Penn">
<meta property="og:description" content="Java的事件机制一般都是通过观察者模式实现的(?).观察者模式就必须有观察者和被观察者.Mina里里面,再一开始acceptor = new NioSocketAcceptor(NetConfig.MINA_PROCESSOR);就在上面的父类AbstractIoService中注册了观察者:1234567891011121314151617181920212223242526272829303">
<meta property="og:locale" content="zh-Hans">
<meta property="og:updated_time" content="2019-05-28T03:57:49.261Z">
<meta name="twitter:card" content="summary">
<meta name="twitter:title" content="Mina中的观察者模式">
<meta name="twitter:description" content="Java的事件机制一般都是通过观察者模式实现的(?).观察者模式就必须有观察者和被观察者.Mina里里面,再一开始acceptor = new NioSocketAcceptor(NetConfig.MINA_PROCESSOR);就在上面的父类AbstractIoService中注册了观察者:1234567891011121314151617181920212223242526272829303">



<script type="text/javascript" id="hexo.configurations">
  var NexT = window.NexT || {};
  var CONFIG = {
    root: '/',
    scheme: 'Pisces',
    version: '5.1.4',
    sidebar: {"position":"left","display":"post","offset":12,"b2t":false,"scrollpercent":false,"onmobile":true},
    fancybox: true,
    tabs: true,
    motion: {"enable":true,"async":false,"transition":{"post_block":"fadeIn","post_header":"slideDownIn","post_body":"slideDownIn","coll_header":"slideLeftIn","sidebar":"slideUpIn"}},
    duoshuo: {
      userId: '0',
      author: '博主'
    },
    algolia: {
      applicationID: '',
      apiKey: '',
      indexName: '',
      hits: {"per_page":10},
      labels: {"input_placeholder":"Search for Posts","hits_empty":"We didn't find any results for the search: ${query}","hits_stats":"${hits} results found in ${time} ms"}
    }
  };
</script>



  <link rel="canonical" href="http://yoursite.com/2019/05/28/Mina中的观察者模式/">





  <title>Mina中的观察者模式 | Penn</title>
  








</head>

<body itemscope="" itemtype="http://schema.org/WebPage" lang="zh-Hans">

  
  
    
  

  <div class="container sidebar-position-left page-post-detail">
    <div class="headband"></div>

    <header id="header" class="header" itemscope="" itemtype="http://schema.org/WPHeader">
      <div class="header-inner"><div class="site-brand-wrapper">
  <div class="site-meta ">
    

    <div class="custom-logo-site-title">
      <a href="/" class="brand" rel="start">
        <span class="logo-line-before"><i></i></span>
        <span class="site-title">Penn</span>
        <span class="logo-line-after"><i></i></span>
      </a>
    </div>
      
        <p class="site-subtitle">my blog</p>
      
  </div>

  <div class="site-nav-toggle">
    <button>
      <span class="btn-bar"></span>
      <span class="btn-bar"></span>
      <span class="btn-bar"></span>
    </button>
  </div>
</div>

<nav class="site-nav">
  

  
    <ul id="menu" class="menu">
      
        
        <li class="menu-item menu-item-home">
          <a href="/" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-home"></i> <br>
            
            首页
          </a>
        </li>
      
        
        <li class="menu-item menu-item-about">
          <a href="/about/" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-user"></i> <br>
            
            关于
          </a>
        </li>
      
        
        <li class="menu-item menu-item-tags">
          <a href="/tags/" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-tags"></i> <br>
            
            标签
          </a>
        </li>
      
        
        <li class="menu-item menu-item-categories">
          <a href="/categories/" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-th"></i> <br>
            
            分类
          </a>
        </li>
      
        
        <li class="menu-item menu-item-archives">
          <a href="/archives/" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-archive"></i> <br>
            
            归档
          </a>
        </li>
      
        
        <li class="menu-item menu-item-schedule">
          <a href="/schedule/" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-calendar"></i> <br>
            
            日程表
          </a>
        </li>
      
        
        <li class="menu-item menu-item-sitemap">
          <a href="/sitemap.xml" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-sitemap"></i> <br>
            
            站点地图
          </a>
        </li>
      
        
        <li class="menu-item menu-item-commonweal">
          <a href="/404/" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-heartbeat"></i> <br>
            
            公益404
          </a>
        </li>
      

      
    </ul>
  

  
</nav>



 </div>
    </header>

    <main id="main" class="main">
      <div class="main-inner">
        <div class="content-wrap">
          <div id="content" class="content">
            

  <div id="posts" class="posts-expand">
    

  

  
  
  

  <article class="post post-type-normal" itemscope="" itemtype="http://schema.org/Article">
  
  
  
  <div class="post-block">
    <link itemprop="mainEntityOfPage" href="http://yoursite.com/2019/05/28/Mina中的观察者模式/">

    <span hidden itemprop="author" itemscope="" itemtype="http://schema.org/Person">
      <meta itemprop="name" content="Shen">
      <meta itemprop="description" content="">
      <meta itemprop="image" content="/images/saiki.jpeg">
    </span>

    <span hidden itemprop="publisher" itemscope="" itemtype="http://schema.org/Organization">
      <meta itemprop="name" content="Penn">
    </span>

    
      <header class="post-header">

        
        
          <h1 class="post-title" itemprop="name headline">Mina中的观察者模式</h1>
        

        <div class="post-meta">
          <span class="post-time">
            
              <span class="post-meta-item-icon">
                <i class="fa fa-calendar-o"></i>
              </span>
              
                <span class="post-meta-item-text">发表于</span>
              
              <time title="创建于" itemprop="dateCreated datePublished" datetime="2019-05-28T11:57:49+08:00">
                2019-05-28
              </time>
            

            

            
          </span>

          
            <span class="post-category">
            
              <span class="post-meta-divider">|</span>
            
              <span class="post-meta-item-icon">
                <i class="fa fa-folder-o"></i>
              </span>
              
                <span class="post-meta-item-text">分类于</span>
              
              
                <span itemprop="about" itemscope="" itemtype="http://schema.org/Thing">
                  <a href="/categories/技术/" itemprop="url" rel="index">
                    <span itemprop="name">技术</span>
                  </a>
                </span>

                
                
              
            </span>
          

          
            
          

          
          
             <span id="/2019/05/28/Mina中的观察者模式/" class="leancloud_visitors" data-flag-title="Mina中的观察者模式">
               <span class="post-meta-divider">|</span>
               <span class="post-meta-item-icon">
                 <i class="fa fa-eye"></i>
               </span>
               
                 <span class="post-meta-item-text">阅读次数&#58;</span>
               
                 <span class="leancloud-visitors-count"></span>
             </span>
          

          

          

          

        </div>
      </header>
    

    
    
    
    <div class="post-body" itemprop="articleBody">

      
      

      
        <p>Java的事件机制一般都是通过观察者模式实现的(?).<br>观察者模式就必须有观察者和被观察者.<br>Mina里里面,再一开始<br><code>acceptor = new NioSocketAcceptor(NetConfig.MINA_PROCESSOR);</code><br>就在上面的父类<code>AbstractIoService</code>中注册了观察者:<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br></pre></td><td class="code"><pre><span class="line">protected AbstractIoService(IoSessionConfig sessionConfig, Executor executor) &#123;</span><br><span class="line">        if (sessionConfig == null) &#123;</span><br><span class="line">            throw new IllegalArgumentException(&quot;sessionConfig&quot;);</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        if (getTransportMetadata() == null) &#123;</span><br><span class="line">            throw new IllegalArgumentException(&quot;TransportMetadata&quot;);</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        if (!getTransportMetadata().getSessionConfigType().isAssignableFrom(sessionConfig.getClass())) &#123;</span><br><span class="line">            throw new IllegalArgumentException(&quot;sessionConfig type: &quot; + sessionConfig.getClass() + &quot; (expected: &quot;</span><br><span class="line">                    + getTransportMetadata().getSessionConfigType() + &quot;)&quot;);</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line"></span><br><span class="line"></span><br><span class="line">//---------------这里!!!--------------</span><br><span class="line">        // Create the listeners, and add a first listener : a activation listener</span><br><span class="line">        // for this service, which will give information on the service state.</span><br><span class="line">        listeners = new IoServiceListenerSupport(this);</span><br><span class="line">        listeners.add(serviceActivationListener);</span><br><span class="line"></span><br><span class="line"></span><br><span class="line"></span><br><span class="line">        // Stores the given session configuration</span><br><span class="line">        this.sessionConfig = sessionConfig;</span><br><span class="line"></span><br><span class="line">        // Make JVM load the exception monitor before some transports</span><br><span class="line">        // change the thread context class loader.</span><br><span class="line">        ExceptionMonitor.getInstance();</span><br><span class="line"></span><br><span class="line">        if (executor == null) &#123;</span><br><span class="line">            this.executor = Executors.newCachedThreadPool();</span><br><span class="line">            createdExecutor = true;</span><br><span class="line">        &#125; else &#123;</span><br><span class="line">            this.executor = executor;</span><br><span class="line">            createdExecutor = false;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        threadName = getClass().getSimpleName() + &apos;-&apos; + id.incrementAndGet();</span><br><span class="line">    &#125;</span><br></pre></td></tr></table></figure></p>
<p>默认的listener再同一个类中类中定义为:<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br></pre></td><td class="code"><pre><span class="line">private final IoServiceListener serviceActivationListener = new IoServiceListener() &#123;</span><br><span class="line">       public void serviceActivated(IoService service) &#123;</span><br><span class="line">           // Update lastIoTime.</span><br><span class="line">           AbstractIoService s = (AbstractIoService) service;</span><br><span class="line">           IoServiceStatistics _stats = s.getStatistics();</span><br><span class="line">           _stats.setLastReadTime(s.getActivationTime());</span><br><span class="line">           _stats.setLastWriteTime(s.getActivationTime());</span><br><span class="line">           _stats.setLastThroughputCalculationTime(s.getActivationTime());</span><br><span class="line"></span><br><span class="line">       &#125;</span><br><span class="line"></span><br><span class="line">       public void serviceDeactivated(IoService service) throws Exception &#123;</span><br><span class="line">           // Empty handler</span><br><span class="line">       &#125;</span><br><span class="line"></span><br><span class="line">       public void serviceIdle(IoService service, IdleStatus idleStatus) throws Exception &#123;</span><br><span class="line">           // Empty handler</span><br><span class="line">       &#125;</span><br><span class="line"></span><br><span class="line">       public void sessionCreated(IoSession session) throws Exception &#123;</span><br><span class="line">           // Empty handler</span><br><span class="line">       &#125;</span><br><span class="line"></span><br><span class="line">       public void sessionClosed(IoSession session) throws Exception &#123;</span><br><span class="line">           // Empty handler</span><br><span class="line">       &#125;</span><br><span class="line"></span><br><span class="line">       public void sessionDestroyed(IoSession session) throws Exception &#123;</span><br><span class="line">           // Empty handler</span><br><span class="line">       &#125;</span><br><span class="line">   &#125;;</span><br></pre></td></tr></table></figure></p>
<p>有了观察者,现在就是考虑如何再事件发生时通知观察者了.</p>
<p>还是在<code>AbstractPollingIoProcessor</code>中的<code>Processor</code>内部类的run方法中,(mina2.0.17)<br>会调用<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line">// Manage newly created session first</span><br><span class="line">               if(handleNewSessions() == 0) &#123;</span><br></pre></td></tr></table></figure></p>
<p>当有新的session连接时,新的session会加到newSessions里面,<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br></pre></td><td class="code"><pre><span class="line">private int handleNewSessions() &#123;</span><br><span class="line">           int addedSessions = 0;</span><br><span class="line"></span><br><span class="line">           for (S session = newSessions.poll(); session != null; session = newSessions.poll()) &#123;</span><br><span class="line">               if (addNow(session)) &#123;</span><br><span class="line">                   // A new session has been created</span><br><span class="line">                   addedSessions++;</span><br><span class="line">               &#125;</span><br><span class="line">           &#125;</span><br><span class="line"></span><br><span class="line">           return addedSessions;</span><br><span class="line">       &#125;</span><br></pre></td></tr></table></figure></p>
<p>上面方法调用addNow(S)<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br></pre></td><td class="code"><pre><span class="line">private boolean addNow(S session) &#123;</span><br><span class="line">           boolean registered = false;</span><br><span class="line"></span><br><span class="line">           try &#123;</span><br><span class="line">               init(session);</span><br><span class="line">               registered = true;</span><br><span class="line"></span><br><span class="line">               // Build the filter chain of this session.</span><br><span class="line">               IoFilterChainBuilder chainBuilder = session.getService().getFilterChainBuilder();</span><br><span class="line">               chainBuilder.buildFilterChain(session.getFilterChain());</span><br><span class="line"></span><br><span class="line">               // DefaultIoFilterChain.CONNECT_FUTURE is cleared inside here</span><br><span class="line">               // in AbstractIoFilterChain.fireSessionOpened().</span><br><span class="line">               // Propagate the SESSION_CREATED event up to the chain</span><br><span class="line">               IoServiceListenerSupport listeners = ((AbstractIoService) session.getService()).getListeners();</span><br><span class="line">               listeners.fireSessionCreated(session);</span><br><span class="line">           &#125; catch (Exception e) &#123;</span><br><span class="line">               ExceptionMonitor.getInstance().exceptionCaught(e);</span><br><span class="line"></span><br><span class="line">               try &#123;</span><br><span class="line">                   destroy(session);</span><br><span class="line">               &#125; catch (Exception e1) &#123;</span><br><span class="line">                   ExceptionMonitor.getInstance().exceptionCaught(e1);</span><br><span class="line">               &#125; finally &#123;</span><br><span class="line">                   registered = false;</span><br><span class="line">               &#125;</span><br><span class="line">           &#125;</span><br><span class="line"></span><br><span class="line">           return registered;</span><br><span class="line">       &#125;</span><br></pre></td></tr></table></figure></p>
<p>里面<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line">IoServiceListenerSupport listeners = ((AbstractIoService) session.getService()).getListeners();</span><br><span class="line">              listeners.fireSessionCreated(session);</span><br></pre></td></tr></table></figure></p>
<p>listeners就是在开头代码里赋值的,而listener也在开头代码里注册了进去.<br>在它的<code>fireSessionCreated</code>方法里面<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br></pre></td><td class="code"><pre><span class="line">public void fireSessionCreated(IoSession session) &#123;</span><br><span class="line">        boolean firstSession = false;</span><br><span class="line"></span><br><span class="line">        if (session.getService() instanceof IoConnector) &#123;</span><br><span class="line">            synchronized (managedSessions) &#123;</span><br><span class="line">                firstSession = managedSessions.isEmpty();</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        // If already registered, ignore.</span><br><span class="line">        if (managedSessions.putIfAbsent(session.getId(), session) != null) &#123;</span><br><span class="line">            return;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        // If the first connector session, fire a virtual service activation event.</span><br><span class="line">        if (firstSession) &#123;</span><br><span class="line">            fireServiceActivated();</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        // Fire session events.</span><br><span class="line">        IoFilterChain filterChain = session.getFilterChain();</span><br><span class="line">        filterChain.fireSessionCreated();</span><br><span class="line">        filterChain.fireSessionOpened();</span><br><span class="line"></span><br><span class="line">        int managedSessionCount = managedSessions.size();</span><br><span class="line"></span><br><span class="line">        if (managedSessionCount &gt; largestManagedSessionCount) &#123;</span><br><span class="line">            largestManagedSessionCount = managedSessionCount;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        cumulativeManagedSessionCount.incrementAndGet();</span><br><span class="line"></span><br><span class="line">        // Fire listener events.</span><br><span class="line">        for (IoServiceListener l : listeners) &#123;</span><br><span class="line">            try &#123;</span><br><span class="line">                l.sessionCreated(session);</span><br><span class="line">            &#125; catch (Exception e) &#123;</span><br><span class="line">                ExceptionMonitor.getInstance().exceptionCaught(e);</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br></pre></td></tr></table></figure></p>
<p>可以看到<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br></pre></td><td class="code"><pre><span class="line">// Fire session events.</span><br><span class="line">     IoFilterChain filterChain = session.getFilterChain();</span><br><span class="line">     filterChain.fireSessionCreated();</span><br><span class="line">     filterChain.fireSessionOpened();</span><br></pre></td></tr></table></figure></p>
<p>sessionCreated、sessionOpened事件就是这么触发的.<br>sessionIdle事件在<code>mina的IoEvent</code>里面写了.<br>剩下的messageReceive合messageSent还是回到<code>AbstractPollingIoProcessor</code>里面,在<code>run</code>方法里面<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br></pre></td><td class="code"><pre><span class="line"></span><br><span class="line">// Now, if we have had some incoming or outgoing events,</span><br><span class="line">// deal with them</span><br><span class="line">if (selected &gt; 0) &#123;</span><br><span class="line">    // LOG.debug(&quot;Processing ...&quot;); // This log hurts one of</span><br><span class="line">    // the MDCFilter test...</span><br><span class="line">    process();</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure></p>
<p><code>process()</code>方法:<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br></pre></td><td class="code"><pre><span class="line">private void process() throws Exception &#123;</span><br><span class="line">            for (Iterator&lt;S&gt; i = selectedSessions(); i.hasNext();) &#123;</span><br><span class="line">                S session = i.next();</span><br><span class="line">                process(session);</span><br><span class="line">                i.remove();</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br></pre></td></tr></table></figure></p>
<p>继续看<code>process(S session)</code>方法<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br></pre></td><td class="code"><pre><span class="line">/**</span><br><span class="line">        * Deal with session ready for the read or write operations, or both.</span><br><span class="line">        */</span><br><span class="line">       private void process(S session) &#123;</span><br><span class="line">           // Process Reads</span><br><span class="line">           if (isReadable(session) &amp;&amp; !session.isReadSuspended()) &#123;</span><br><span class="line">               read(session);</span><br><span class="line">           &#125;</span><br><span class="line"></span><br><span class="line">           // Process writes</span><br><span class="line">           if (isWritable(session) &amp;&amp; !session.isWriteSuspended() &amp;&amp; session.setScheduledForFlush(true)) &#123;</span><br><span class="line">               // add the session to the queue, if it&apos;s not already there</span><br><span class="line">               flushingSessions.add(session);</span><br><span class="line">           &#125;</span><br><span class="line">       &#125;</span><br></pre></td></tr></table></figure></p>
<p>#####1.messageReceived<br>就是在上面代码的<code>read</code>方法里面<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br></pre></td><td class="code"><pre><span class="line">private void read(S session) &#123;</span><br><span class="line">        IoSessionConfig config = session.getConfig();</span><br><span class="line">        int bufferSize = config.getReadBufferSize();</span><br><span class="line">        IoBuffer buf = IoBuffer.allocate(bufferSize);</span><br><span class="line"></span><br><span class="line">        final boolean hasFragmentation = session.getTransportMetadata().hasFragmentation();</span><br><span class="line"></span><br><span class="line">        try &#123;</span><br><span class="line">            int readBytes = 0;</span><br><span class="line">            int ret;</span><br><span class="line"></span><br><span class="line">            try &#123;</span><br><span class="line">                if (hasFragmentation) &#123;</span><br><span class="line"></span><br><span class="line">                    while ((ret = read(session, buf)) &gt; 0) &#123;</span><br><span class="line">                        readBytes += ret;</span><br><span class="line"></span><br><span class="line">                        if (!buf.hasRemaining()) &#123;</span><br><span class="line">                            break;</span><br><span class="line">                        &#125;</span><br><span class="line">                    &#125;</span><br><span class="line">                &#125; else &#123;</span><br><span class="line">                    ret = read(session, buf);</span><br><span class="line"></span><br><span class="line">                    if (ret &gt; 0) &#123;</span><br><span class="line">                        readBytes = ret;</span><br><span class="line">                    &#125;</span><br><span class="line">                &#125;</span><br><span class="line">            &#125; finally &#123;</span><br><span class="line">                buf.flip();</span><br><span class="line">            &#125;</span><br><span class="line"></span><br><span class="line">            if (readBytes &gt; 0) &#123;</span><br><span class="line">                IoFilterChain filterChain = session.getFilterChain();</span><br><span class="line">                filterChain.fireMessageReceived(buf);</span><br><span class="line">                buf = null;</span><br><span class="line"></span><br><span class="line">                if (hasFragmentation) &#123;</span><br><span class="line">                    if (readBytes &lt;&lt; 1 &lt; config.getReadBufferSize()) &#123;</span><br><span class="line">                        session.decreaseReadBufferSize();</span><br><span class="line">                    &#125; else if (readBytes == config.getReadBufferSize()) &#123;</span><br><span class="line">                        session.increaseReadBufferSize();</span><br><span class="line">                    &#125;</span><br><span class="line">                &#125;</span><br><span class="line">            &#125; else &#123;</span><br><span class="line">                // release temporary buffer when read nothing</span><br><span class="line">                buf.free(); </span><br><span class="line">            &#125;</span><br><span class="line"></span><br><span class="line">            if (ret &lt; 0) &#123;</span><br><span class="line">                IoFilterChain filterChain = session.getFilterChain();</span><br><span class="line">                filterChain.fireInputClosed();</span><br><span class="line">            &#125;</span><br><span class="line">        &#125; catch (Exception e) &#123;</span><br><span class="line">            if ((e instanceof IOException) &amp;&amp;</span><br><span class="line">                (!(e instanceof PortUnreachableException)</span><br><span class="line">                        || !AbstractDatagramSessionConfig.class.isAssignableFrom(config.getClass())</span><br><span class="line">                        || ((AbstractDatagramSessionConfig) config).isCloseOnPortUnreachable())) &#123;</span><br><span class="line">                scheduleRemove(session);</span><br><span class="line">            &#125;</span><br><span class="line"></span><br><span class="line">            IoFilterChain filterChain = session.getFilterChain();</span><br><span class="line">            filterChain.fireExceptionCaught(e);</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br></pre></td></tr></table></figure></p>
<p>#####2.messageSent<br>在上面<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line">加到`flushingSessions`里面.</span><br><span class="line">在`run()`方法里面</span><br></pre></td></tr></table></figure></p>
<p>   // Write the pending requests<br>                    long currentTime = System.currentTimeMillis();<br>                    flush(currentTime);<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">flush</span><br></pre></td></tr></table></figure></p>
<p>private void flush(long currentTime) {<br>            if (flushingSessions.isEmpty()) {<br>                return;<br>            }</p>
<pre><code>    do {
        S session = flushingSessions.poll(); // the same one with
                                             // firstSession

        if (session == null) {
            // Just in case ... It should not happen.
            break;
        }

        // Reset the Schedule for flush flag for this session,
        // as we are flushing it now
        session.unscheduledForFlush();

        SessionState state = getState(session);

        switch (state) {
        case OPENED:
            try {
                boolean flushedAll = flushNow(session, currentTime);

                if (flushedAll &amp;&amp; !session.getWriteRequestQueue().isEmpty(session)
                        &amp;&amp; !session.isScheduledForFlush()) {
                    scheduleFlush(session);
                }
            } catch (Exception e) {
                scheduleRemove(session);
                session.closeNow();
                IoFilterChain filterChain = session.getFilterChain();
                filterChain.fireExceptionCaught(e);
            }

            break;

        case CLOSING:
            // Skip if the channel is already closed.
            break;

        case OPENING:
            // Retry later if session is not yet fully initialized.
            // (In case that Session.write() is called before addSession()
            // is processed)
            scheduleFlush(session);
            return;

        default:
            throw new IllegalStateException(String.valueOf(state));
        }

    } while (!flushingSessions.isEmpty());
}
</code></pre><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">flushNow</span><br></pre></td></tr></table></figure>
<p> private boolean flushNow(S session, long currentTime) {<br>            if (!session.isConnected()) {<br>                scheduleRemove(session);<br>                return false;<br>            }</p>
<pre><code>    final boolean hasFragmentation = session.getTransportMetadata().hasFragmentation();

    final WriteRequestQueue writeRequestQueue = session.getWriteRequestQueue();

    // Set limitation for the number of written bytes for read-write
    // fairness. I used maxReadBufferSize * 3 / 2, which yields best
    // performance in my experience while not breaking fairness much.
    final int maxWrittenBytes = session.getConfig().getMaxReadBufferSize()
            + (session.getConfig().getMaxReadBufferSize() &gt;&gt;&gt; 1);
    int writtenBytes = 0;
    WriteRequest req = null;

    try {
        // Clear OP_WRITE
        setInterestedInWrite(session, false);

        do {
            // Check for pending writes.
            req = session.getCurrentWriteRequest();

            if (req == null) {
                req = writeRequestQueue.poll(session);

                if (req == null) {
                    break;
                }

                session.setCurrentWriteRequest(req);
            }

            int localWrittenBytes;
            Object message = req.getMessage();

            if (message instanceof IoBuffer) {
                localWrittenBytes = writeBuffer(session, req, hasFragmentation, maxWrittenBytes - writtenBytes,
                        currentTime);

                if ((localWrittenBytes &gt; 0) &amp;&amp; ((IoBuffer) message).hasRemaining()) {
                    // the buffer isn&apos;t empty, we re-interest it in writing
                    setInterestedInWrite(session, true);

                    return false;
                }
            } else if (message instanceof FileRegion) {
                localWrittenBytes = writeFile(session, req, hasFragmentation, maxWrittenBytes - writtenBytes,
                        currentTime);

                // Fix for Java bug on Linux
                // http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=5103988
                // If there&apos;s still data to be written in the FileRegion,
                // return 0 indicating that we need
                // to pause until writing may resume.
                if ((localWrittenBytes &gt; 0) &amp;&amp; (((FileRegion) message).getRemainingBytes() &gt; 0)) {
                    setInterestedInWrite(session, true);

                    return false;
                }
            } else {
                throw new IllegalStateException(&quot;Don&apos;t know how to handle message of type &apos;&quot;
                        + message.getClass().getName() + &quot;&apos;.  Are you missing a protocol encoder?&quot;);
            }

            if (localWrittenBytes == 0) {

                // Kernel buffer is full.
                if (!req.equals(AbstractIoSession.MESSAGE_SENT_REQUEST)) {
                    setInterestedInWrite(session, true);
                    return false;
                }
            } else {
                writtenBytes += localWrittenBytes;

                if (writtenBytes &gt;= maxWrittenBytes) {
                    // Wrote too much
                    scheduleFlush(session);
                    return false;
                }
            }

            if (message instanceof IoBuffer) {
                ((IoBuffer) message).free();
            }
        } while (writtenBytes &lt; maxWrittenBytes);
    } catch (Exception e) {
        if (req != null) {
            req.getFuture().setException(e);
        }

        IoFilterChain filterChain = session.getFilterChain();
        filterChain.fireExceptionCaught(e);
        return false;
    }

    return true;
}
</code></pre><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">其中</span><br></pre></td></tr></table></figure>
<p>else if (message instanceof FileRegion) {<br>                        localWrittenBytes = writeFile(session, req, hasFragmentation, maxWrittenBytes - writtenBytes,<br>                                currentTime);<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">writeFile</span><br></pre></td></tr></table></figure></p>
<p>private int writeFile(S session, WriteRequest req, boolean hasFragmentation, int maxLength, long currentTime)<br>                throws Exception {<br>            int localWrittenBytes;<br>            FileRegion region = (FileRegion) req.getMessage();</p>
<pre><code>    if (region.getRemainingBytes() &gt; 0) {
        int length;

        if (hasFragmentation) {
            length = (int) Math.min(region.getRemainingBytes(), maxLength);
        } else {
            length = (int) Math.min(Integer.MAX_VALUE, region.getRemainingBytes());
        }

        localWrittenBytes = transferFile(session, region, length);
        region.update(localWrittenBytes);
    } else {
        localWrittenBytes = 0;
    }

    session.increaseWrittenBytes(localWrittenBytes, currentTime);

    if ((region.getRemainingBytes() &lt;= 0) || (!hasFragmentation &amp;&amp; (localWrittenBytes != 0))) {
        fireMessageSent(session, req);
    }

    return localWrittenBytes;
}
</code></pre><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">里面</span><br></pre></td></tr></table></figure>
<p>if ((region.getRemainingBytes() &lt;= 0) || (!hasFragmentation &amp;&amp; (localWrittenBytes != 0))) {<br>                fireMessageSent(session, req);<br>            }<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br></pre></td><td class="code"><pre><span class="line">就会出发messageSent了</span><br><span class="line"></span><br><span class="line"></span><br><span class="line"></span><br><span class="line">###小结</span><br><span class="line">可以看到,这些事件都是由`AbstractPollingIoProcessor`的内部类`Processor`的`run`方法驱动的.而`run`方法就是在</span><br></pre></td></tr></table></figure></p>
<p>private void startupProcessor() {<br>        Processor processor = processorRef.get();</p>
<pre><code>    if (processor == null) {
        processor = new Processor();

        if (processorRef.compareAndSet(null, processor)) {
            executor.execute(new NamePreservingRunnable(processor, threadName));
        }
    }

    // Just stop the select() and start it again, so that the processor
    // can be activated immediately.
    wakeup();
}
</code></pre><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">的时候开始运行的,有三个方法调用`startupProcessor`:</span><br></pre></td></tr></table></figure>
<p>1.AbstractPollingIoProcessor.add()<br>2.AbstractPollingIoProcessor.remove()<br>3.AbstractPollingIoProcessor.dispose()<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br></pre></td><td class="code"><pre><span class="line">所以,在第一个连接过来的时候,`executor`里面的线程就在跑`run`方法了.</span><br><span class="line"></span><br><span class="line"></span><br><span class="line">--------</span><br><span class="line">接一开始的`newSessions`,为什么有新的连接,就会有·session·加进`newSessions`里面呢?</span><br><span class="line"></span><br><span class="line">在mina的服务器代码里面,启动mina时需要调用`acceptor.bind();`</span><br><span class="line">追踪`bind`方法</span><br></pre></td></tr></table></figure></p>
<p> public final void bind(Iterable&lt;? extends SocketAddress&gt; localAddresses) throws IOException {<br>        if (isDisposing()) {<br>            throw new IllegalStateException(“The Accpetor disposed is being disposed.”);<br>        }</p>
<pre><code>    if (localAddresses == null) {
        throw new IllegalArgumentException(&quot;localAddresses&quot;);
    }

    List&lt;SocketAddress&gt; localAddressesCopy = new ArrayList&lt;SocketAddress&gt;();

    for (SocketAddress a : localAddresses) {
        checkAddressType(a);
        localAddressesCopy.add(a);
    }

    if (localAddressesCopy.isEmpty()) {
        throw new IllegalArgumentException(&quot;localAddresses is empty.&quot;);
    }

    boolean activate = false;
    synchronized (bindLock) {
        synchronized (boundAddresses) {
            if (boundAddresses.isEmpty()) {
                activate = true;
            }
        }

        if (getHandler() == null) {
            throw new IllegalStateException(&quot;handler is not set.&quot;);
        }

        try {
            Set&lt;SocketAddress&gt; addresses = bindInternal(localAddressesCopy);

            synchronized (boundAddresses) {
                boundAddresses.addAll(addresses);
            }
        } catch (IOException e) {
            throw e;
        } catch (RuntimeException e) {
            throw e;
        } catch (Exception e) {
            throw new RuntimeIoException(&quot;Failed to bind to: &quot; + getLocalAddresses(), e);
        }
    }

    if (activate) {
        getListeners().fireServiceActivated();
    }
}
</code></pre><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line">有一行</span><br><span class="line">``` Set&lt;SocketAddress&gt; addresses = bindInternal(localAddressesCopy);`</span><br></pre></td></tr></table></figure>
<p>继续追踪<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br></pre></td><td class="code"><pre><span class="line">protected final Set&lt;SocketAddress&gt; bindInternal(List&lt;? extends SocketAddress&gt; localAddresses) throws Exception &#123;</span><br><span class="line">       // Create a bind request as a Future operation. When the selector</span><br><span class="line">       // have handled the registration, it will signal this future.</span><br><span class="line">       AcceptorOperationFuture request = new AcceptorOperationFuture(localAddresses);</span><br><span class="line"></span><br><span class="line">       // adds the Registration request to the queue for the Workers</span><br><span class="line">       // to handle</span><br><span class="line">       registerQueue.add(request);</span><br><span class="line"></span><br><span class="line">       // creates the Acceptor instance and has the local</span><br><span class="line">       // executor kick it off.</span><br><span class="line">       startupAcceptor();</span><br><span class="line"></span><br><span class="line">       // As we just started the acceptor, we have to unblock the select()</span><br><span class="line">       // in order to process the bind request we just have added to the</span><br><span class="line">       // registerQueue.</span><br><span class="line">       try &#123;</span><br><span class="line">           lock.acquire();</span><br><span class="line"></span><br><span class="line">           // Wait a bit to give a chance to the Acceptor thread to do the select()</span><br><span class="line">           Thread.sleep(10);</span><br><span class="line">           wakeup();</span><br><span class="line">       &#125; finally &#123;</span><br><span class="line">           lock.release();</span><br><span class="line">       &#125;</span><br><span class="line"></span><br><span class="line">       // Now, we wait until this request is completed.</span><br><span class="line">       request.awaitUninterruptibly();</span><br><span class="line"></span><br><span class="line">       if (request.getException() != null) &#123;</span><br><span class="line">           throw request.getException();</span><br><span class="line">       &#125;</span><br><span class="line"></span><br><span class="line">       // Update the local addresses.</span><br><span class="line">       // setLocalAddresses() shouldn&apos;t be called from the worker thread</span><br><span class="line">       // because of deadlock.</span><br><span class="line">       Set&lt;SocketAddress&gt; newLocalAddresses = new HashSet&lt;SocketAddress&gt;();</span><br><span class="line"></span><br><span class="line">       for (H handle : boundHandles.values()) &#123;</span><br><span class="line">           newLocalAddresses.add(localAddress(handle));</span><br><span class="line">       &#125;</span><br><span class="line"></span><br><span class="line">       return newLocalAddresses;</span><br><span class="line">   &#125;</span><br></pre></td></tr></table></figure></p>
<p>里面调用了<code>startupAcceptor();</code><br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br></pre></td><td class="code"><pre><span class="line">private void startupAcceptor() throws InterruptedException &#123;</span><br><span class="line">        // If the acceptor is not ready, clear the queues</span><br><span class="line">        // TODO : they should already be clean : do we have to do that ?</span><br><span class="line">        if (!selectable) &#123;</span><br><span class="line">            registerQueue.clear();</span><br><span class="line">            cancelQueue.clear();</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        // start the acceptor if not already started</span><br><span class="line">        Acceptor acceptor = acceptorRef.get();</span><br><span class="line"></span><br><span class="line">        if (acceptor == null) &#123;</span><br><span class="line">            lock.acquire();</span><br><span class="line">            acceptor = new Acceptor();</span><br><span class="line"></span><br><span class="line">            if (acceptorRef.compareAndSet(null, acceptor)) &#123;</span><br><span class="line">                executeWorker(acceptor);</span><br><span class="line">            &#125; else &#123;</span><br><span class="line">                lock.release();</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br></pre></td></tr></table></figure></p>
<p>开启了一个线程执行<code>Acceptor</code>的<code>run()</code>方法:<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br></pre></td><td class="code"><pre><span class="line">private void startupAcceptor() throws InterruptedException &#123;</span><br><span class="line">        // If the acceptor is not ready, clear the queues</span><br><span class="line">        // TODO : they should already be clean : do we have to do that ?</span><br><span class="line">        if (!selectable) &#123;</span><br><span class="line">            registerQueue.clear();</span><br><span class="line">            cancelQueue.clear();</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        // start the acceptor if not already started</span><br><span class="line">        Acceptor acceptor = acceptorRef.get();</span><br><span class="line"></span><br><span class="line">        if (acceptor == null) &#123;</span><br><span class="line">            lock.acquire();</span><br><span class="line">            acceptor = new Acceptor();</span><br><span class="line"></span><br><span class="line">            if (acceptorRef.compareAndSet(null, acceptor)) &#123;</span><br><span class="line">                executeWorker(acceptor);</span><br><span class="line">            &#125; else &#123;</span><br><span class="line">                lock.release();</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br></pre></td></tr></table></figure></p>
<p>可以看到<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br></pre></td><td class="code"><pre><span class="line">if (selected &gt; 0) &#123;</span><br><span class="line">                        // We have some connection request, let&apos;s process</span><br><span class="line">                        // them here.</span><br><span class="line">                        processHandles(selectedHandles());</span><br><span class="line">                    &#125;</span><br></pre></td></tr></table></figure></p>
<p><code>processHandles</code>方法:<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br></pre></td><td class="code"><pre><span class="line">private void processHandles(Iterator&lt;H&gt; handles) throws Exception &#123;</span><br><span class="line">            while (handles.hasNext()) &#123;</span><br><span class="line">                H handle = handles.next();</span><br><span class="line">                handles.remove();</span><br><span class="line"></span><br><span class="line">                // Associates a new created connection to a processor,</span><br><span class="line">                // and get back a session</span><br><span class="line">                S session = accept(processor, handle);</span><br><span class="line"></span><br><span class="line">                if (session == null) &#123;</span><br><span class="line">                    continue;</span><br><span class="line">                &#125;</span><br><span class="line"></span><br><span class="line">                initSession(session, null, null);</span><br><span class="line"></span><br><span class="line">                // add the session to the SocketIoProcessor</span><br><span class="line">                session.getProcessor().add(session);</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br></pre></td></tr></table></figure></p>
<p>这个方法里面调用了三个方法:</p>
<p>#####1.    S session = accept(processor, handle);<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br></pre></td><td class="code"><pre><span class="line">protected NioSession accept(IoProcessor&lt;NioSession&gt; processor, ServerSocketChannel handle) throws Exception &#123;</span><br><span class="line"></span><br><span class="line">        SelectionKey key = null;</span><br><span class="line"></span><br><span class="line">        if (handle != null) &#123;</span><br><span class="line">            key = handle.keyFor(selector);</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        if ((key == null) || (!key.isValid()) || (!key.isAcceptable())) &#123;</span><br><span class="line">            return null;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        // accept the connection from the client</span><br><span class="line">        SocketChannel ch = handle.accept();</span><br><span class="line"></span><br><span class="line">        if (ch == null) &#123;</span><br><span class="line">            return null;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        return new NioSocketSession(this, processor, ch);</span><br><span class="line">    &#125;</span><br></pre></td></tr></table></figure></p>
<p>new了一个session并且注册了SocketChannel</p>
<p>#####2. initSession(session, null, null);<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br></pre></td><td class="code"><pre><span class="line">protected final void initSession(IoSession session, IoFuture future, IoSessionInitializer sessionInitializer) &#123;</span><br><span class="line">    // Update lastIoTime if needed.</span><br><span class="line">    if (stats.getLastReadTime() == 0) &#123;</span><br><span class="line">        stats.setLastReadTime(getActivationTime());</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    if (stats.getLastWriteTime() == 0) &#123;</span><br><span class="line">        stats.setLastWriteTime(getActivationTime());</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    // Every property but attributeMap should be set now.</span><br><span class="line">    // Now initialize the attributeMap.  The reason why we initialize</span><br><span class="line">    // the attributeMap at last is to make sure all session properties</span><br><span class="line">    // such as remoteAddress are provided to IoSessionDataStructureFactory.</span><br><span class="line">    try &#123;</span><br><span class="line">        ((AbstractIoSession) session).setAttributeMap(session.getService().getSessionDataStructureFactory()</span><br><span class="line">                .getAttributeMap(session));</span><br><span class="line">    &#125; catch (IoSessionInitializationException e) &#123;</span><br><span class="line">        throw e;</span><br><span class="line">    &#125; catch (Exception e) &#123;</span><br><span class="line">        throw new IoSessionInitializationException(&quot;Failed to initialize an attributeMap.&quot;, e);</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    try &#123;</span><br><span class="line">        ((AbstractIoSession) session).setWriteRequestQueue(session.getService().getSessionDataStructureFactory()</span><br><span class="line">                .getWriteRequestQueue(session));</span><br><span class="line">    &#125; catch (IoSessionInitializationException e) &#123;</span><br><span class="line">        throw e;</span><br><span class="line">    &#125; catch (Exception e) &#123;</span><br><span class="line">        throw new IoSessionInitializationException(&quot;Failed to initialize a writeRequestQueue.&quot;, e);</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    if ((future != null) &amp;&amp; (future instanceof ConnectFuture)) &#123;</span><br><span class="line">        // DefaultIoFilterChain will notify the future. (We support ConnectFuture only for now).</span><br><span class="line">        session.setAttribute(DefaultIoFilterChain.SESSION_CREATED_FUTURE, future);</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    if (sessionInitializer != null) &#123;</span><br><span class="line">        sessionInitializer.initializeSession(session, future);</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    finishSessionInitialization0(session, future);</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure></p>
<p>对<code>session</code>各种初始化</p>
<p>#####3. session.getProcessor().add(session);<br>这里的<code>Processor</code>就是<code>AbstractPollingIoProcessor</code><br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br></pre></td><td class="code"><pre><span class="line">public final void add(S session) &#123;</span><br><span class="line">        if (disposed || disposing) &#123;</span><br><span class="line">            throw new IllegalStateException(&quot;Already disposed.&quot;);</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        // Adds the session to the newSession queue and starts the worker</span><br><span class="line">        newSessions.add(session);</span><br><span class="line">        startupProcessor();</span><br><span class="line">    &#125;</span><br></pre></td></tr></table></figure></p>
<p>这里就解决了上面没有提到的<br>1.将<code>session</code>加到<code>newSessions</code>里面;<br>2.<code>startupProcessor();</code>运行了那个关键的<code>Processor</code>的<code>Run()</code>方法<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br></pre></td><td class="code"><pre><span class="line">private void startupProcessor() &#123;</span><br><span class="line">       Processor processor = processorRef.get();</span><br><span class="line"></span><br><span class="line">       if (processor == null) &#123;</span><br><span class="line">           processor = new Processor();</span><br><span class="line"></span><br><span class="line">           if (processorRef.compareAndSet(null, processor)) &#123;</span><br><span class="line">               executor.execute(new NamePreservingRunnable(processor, threadName));</span><br><span class="line">           &#125;</span><br><span class="line">       &#125;</span><br><span class="line"></span><br><span class="line">       // Just stop the select() and start it again, so that the processor</span><br><span class="line">       // can be activated immediately.</span><br><span class="line">       wakeup();</span><br><span class="line">   &#125;</span><br></pre></td></tr></table></figure></p>

      
    </div>
    
    
    

    

    
      <div>
        <div style="padding: 10px 0; margin: 20px auto; width: 90%; text-align: center;">
  <div>这两个图片是什么?可以扫吗?</div>
  <button id="rewardButton" disable="enable" onclick="var qr = document.getElementById('QR'); if (qr.style.display === 'none') {qr.style.display='block';} else {qr.style.display='none'}">
    <span>打赏</span>
  </button>
  <div id="QR" style="display: none;">

    
      <div id="wechat" style="display: inline-block">
        <img id="wechat_qr" src="/images/wechatPay.png" alt="Shen 微信支付">
        <p>微信支付</p>
      </div>
    

    
      <div id="alipay" style="display: inline-block">
        <img id="alipay_qr" src="/images/aliPay.png" alt="Shen 支付宝">
        <p>支付宝</p>
      </div>
    

    

  </div>
</div>

      </div>
    

    

    <footer class="post-footer">
      
        <div class="post-tags">
          
            <a href="/tags/Java/" rel="tag"># Java</a>
          
            <a href="/tags/mina/" rel="tag"># mina</a>
          
        </div>
      

      
      
      

      
        <div class="post-nav">
          <div class="post-nav-next post-nav-item">
            
              <a href="/2019/05/28/Mina 为每个session构建filterChain/" rel="next" title="Mina 为每个session构建filterChain">
                <i class="fa fa-chevron-left"></i> Mina 为每个session构建filterChain
              </a>
            
          </div>

          <span class="post-nav-divider"></span>

          <div class="post-nav-prev post-nav-item">
            
              <a href="/2019/05/28/MongoDB $type操作符/" rel="prev" title="MongoDB $type操作符">
                MongoDB $type操作符 <i class="fa fa-chevron-right"></i>
              </a>
            
          </div>
        </div>
      

      
      
    </footer>
  </div>
  
  
  
  </article>



    <div class="post-spread">
      
    </div>
  </div>


          </div>
          


          

  
    <div class="comments" id="comments">
      <div id="lv-container" data-id="city" data-uid="MTAyMC80MjA1OS8xODYwNg=="></div>
    </div>

  



        </div>
        
          
  
  <div class="sidebar-toggle">
    <div class="sidebar-toggle-line-wrap">
      <span class="sidebar-toggle-line sidebar-toggle-line-first"></span>
      <span class="sidebar-toggle-line sidebar-toggle-line-middle"></span>
      <span class="sidebar-toggle-line sidebar-toggle-line-last"></span>
    </div>
  </div>

  <aside id="sidebar" class="sidebar">
    
      <div id="sidebar-dimmer"></div>
    
    <div class="sidebar-inner">

      

      

      <section class="site-overview-wrap sidebar-panel sidebar-panel-active">
        <div class="site-overview">
          <div class="site-author motion-element" itemprop="author" itemscope="" itemtype="http://schema.org/Person">
            
              <img class="site-author-image" itemprop="image" src="/images/saiki.jpeg" alt="Shen">
            
              <p class="site-author-name" itemprop="name">Shen</p>
              <p class="site-description motion-element" itemprop="description">Penn's blog</p>
          </div>

          <nav class="site-state motion-element">

            
              <div class="site-state-item site-state-posts">
              
                <a href="/archives/">
              
                  <span class="site-state-item-count">117</span>
                  <span class="site-state-item-name">日志</span>
                </a>
              </div>
            

            
              
              
              <div class="site-state-item site-state-categories">
                <a href="/categories/index.html">
                  <span class="site-state-item-count">1</span>
                  <span class="site-state-item-name">分类</span>
                </a>
              </div>
            

            
              
              
              <div class="site-state-item site-state-tags">
                <a href="/tags/index.html">
                  <span class="site-state-item-count">26</span>
                  <span class="site-state-item-name">标签</span>
                </a>
              </div>
            

          </nav>

          

          

          
          

          
          

          

        </div>
      </section>

      

      

    </div>
  </aside>


        
      </div>
    </main>

    <footer id="footer" class="footer">
      <div class="footer-inner">
        <div class="copyright">&copy; <span itemprop="copyrightYear">2020</span>
  <span class="with-love">
    <i class="fa fa-user"></i>
  </span>
  <span class="author" itemprop="copyrightHolder">Shen</span>

  
</div>


  <div class="powered-by">由 <a class="theme-link" target="_blank" href="https://hexo.io">Hexo</a> 强力驱动</div>



  <span class="post-meta-divider">|</span>



  <div class="theme-info">主题 &mdash; <a class="theme-link" target="_blank" href="https://github.com/iissnan/hexo-theme-next">NexT.Pisces</a> v5.1.4</div>




        







        
      </div>
    </footer>

    
      <div class="back-to-top">
        <i class="fa fa-arrow-up"></i>
        
      </div>
    

    

  </div>

  

<script type="text/javascript">
  if (Object.prototype.toString.call(window.Promise) !== '[object Function]') {
    window.Promise = null;
  }
</script>









  












  
  
    <script type="text/javascript" src="/lib/jquery/index.js?v=2.1.3"></script>
  

  
  
    <script type="text/javascript" src="/lib/fastclick/lib/fastclick.min.js?v=1.0.6"></script>
  

  
  
    <script type="text/javascript" src="/lib/jquery_lazyload/jquery.lazyload.js?v=1.9.7"></script>
  

  
  
    <script type="text/javascript" src="/lib/velocity/velocity.min.js?v=1.2.1"></script>
  

  
  
    <script type="text/javascript" src="/lib/velocity/velocity.ui.min.js?v=1.2.1"></script>
  

  
  
    <script type="text/javascript" src="/lib/fancybox/source/jquery.fancybox.pack.js?v=2.1.5"></script>
  


  


  <script type="text/javascript" src="/js/src/utils.js?v=5.1.4"></script>

  <script type="text/javascript" src="/js/src/motion.js?v=5.1.4"></script>



  
  


  <script type="text/javascript" src="/js/src/affix.js?v=5.1.4"></script>

  <script type="text/javascript" src="/js/src/schemes/pisces.js?v=5.1.4"></script>



  
  <script type="text/javascript" src="/js/src/scrollspy.js?v=5.1.4"></script>
<script type="text/javascript" src="/js/src/post-details.js?v=5.1.4"></script>



  


  <script type="text/javascript" src="/js/src/bootstrap.js?v=5.1.4"></script>



  


  




	





  





  
    <script type="text/javascript">
      (function(d, s) {
        var j, e = d.getElementsByTagName(s)[0];
        if (typeof LivereTower === 'function') { return; }
        j = d.createElement(s);
        j.src = 'https://cdn-city.livere.com/js/embed.dist.js';
        j.async = true;
        e.parentNode.insertBefore(j, e);
      })(document, 'script');
    </script>
  












  





  

  
  <script src="https://cdn1.lncld.net/static/js/av-core-mini-0.6.4.js"></script>
  <script>AV.initialize("Gst71Gv0w5HKejm5H7LuG0XI-gzGzoHsz", "aoFAyHFDKihGTWWhHX0p8byC");</script>
  <script>
    function showTime(Counter) {
      var query = new AV.Query(Counter);
      var entries = [];
      var $visitors = $(".leancloud_visitors");

      $visitors.each(function () {
        entries.push( $(this).attr("id").trim() );
      });

      query.containedIn('url', entries);
      query.find()
        .done(function (results) {
          var COUNT_CONTAINER_REF = '.leancloud-visitors-count';

          if (results.length === 0) {
            $visitors.find(COUNT_CONTAINER_REF).text(0);
            return;
          }

          for (var i = 0; i < results.length; i++) {
            var item = results[i];
            var url = item.get('url');
            var time = item.get('time');
            var element = document.getElementById(url);

            $(element).find(COUNT_CONTAINER_REF).text(time);
          }
          for(var i = 0; i < entries.length; i++) {
            var url = entries[i];
            var element = document.getElementById(url);
            var countSpan = $(element).find(COUNT_CONTAINER_REF);
            if( countSpan.text() == '') {
              countSpan.text(0);
            }
          }
        })
        .fail(function (object, error) {
          console.log("Error: " + error.code + " " + error.message);
        });
    }

    function addCount(Counter) {
      var $visitors = $(".leancloud_visitors");
      var url = $visitors.attr('id').trim();
      var title = $visitors.attr('data-flag-title').trim();
      var query = new AV.Query(Counter);

      query.equalTo("url", url);
      query.find({
        success: function(results) {
          if (results.length > 0) {
            var counter = results[0];
            counter.fetchWhenSave(true);
            counter.increment("time");
            counter.save(null, {
              success: function(counter) {
                var $element = $(document.getElementById(url));
                $element.find('.leancloud-visitors-count').text(counter.get('time'));
              },
              error: function(counter, error) {
                console.log('Failed to save Visitor num, with error message: ' + error.message);
              }
            });
          } else {
            var newcounter = new Counter();
            /* Set ACL */
            var acl = new AV.ACL();
            acl.setPublicReadAccess(true);
            acl.setPublicWriteAccess(true);
            newcounter.setACL(acl);
            /* End Set ACL */
            newcounter.set("title", title);
            newcounter.set("url", url);
            newcounter.set("time", 1);
            newcounter.save(null, {
              success: function(newcounter) {
                var $element = $(document.getElementById(url));
                $element.find('.leancloud-visitors-count').text(newcounter.get('time'));
              },
              error: function(newcounter, error) {
                console.log('Failed to create');
              }
            });
          }
        },
        error: function(error) {
          console.log('Error:' + error.code + " " + error.message);
        }
      });
    }

    $(function() {
      var Counter = AV.Object.extend("Counter");
      if ($('.leancloud_visitors').length == 1) {
        addCount(Counter);
      } else if ($('.post-title-link').length > 1) {
        showTime(Counter);
      }
    });
  </script>



  

  

  
  

  

  

  

</body>
</html>
